// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu, jhkim

#pragma once

#include <fmt/format.h>

#include <experimental/filesystem>
#include <fstream>
#include <memory>
#include <string>
#include <utility>
#include <vector>
#include <unordered_map>
#include <map>
#include <tuple>

#include <boost/exception/diagnostic_information.hpp>
#include "presto/quant/math/moving_average.h"

#include "fastfeature/linear_model_spec.pb.h"
#include "coin2/base/time.h"
#include "coin2/base/log.h"
#include "coin2/exchange/base/feed/update.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/strategy/hooks/pnl_monitor.h"
#include "coin2/strategy/linear_model/logic_factory.h"
#include "coin2/strategy/linear_model/product_strategy.h"
#include "cc/coin2/strategy/linear_model/skip_manager.h"
#include "coin2/strategy/linear_model/util.h"
#include "coin2/strategy/linear_model/util_ban_vu.h"
#include "coin2/strategy/strategy.h"
#include "coin2/strategy/util/config_reporter.h"
#include "coin2/strategy/util/alarm.h"
namespace coin2::strategy::linear_model {

namespace impl {

using ::coin2::exchange::base::market::MarketType;
using ::coin2::exchange::base::feed::FeedUpdate;
using ::coin2::strategy::hooks::PnlMonitor;
using presto::math::TimeWindowMovingAverage;


class LmStrategy : public ::coin2::strategy::IStrategy {
 public:
  typedef typename std::map<
      std::tuple<std::string, std::string, int>, LinearModelWrapper> LinearModelCache;

  std::experimental::filesystem::path lm_strat_json;
  std::experimental::filesystem::path kline_json;
  int64_t timestamp_override;
  int64_t nanosecs_warmedup = 0L;

  LmStrategy(
      const std::experimental::filesystem::path& lm_strat_json,
      const std::experimental::filesystem::path& kline_json,
      int64_t timestamp_override,
      int64_t nanosecs_warmedup = 0L,
      bool simulation = false)
      : lm_strat_json(lm_strat_json),
        kline_json(kline_json),
        timestamp_override(timestamp_override),
        nanosecs_warmedup(nanosecs_warmedup),
        simulation_(simulation),
        last_mtm_update_time_(0L),
        last_book_time_(0L) {}

  void InjectFeed(const coin2::feed::mgr::FeedSubsystemManager* feed_mgr) final {
    std::shared_ptr<::fastfeature::KlineInfoProto> kline_info(new ::fastfeature::KlineInfoProto());
    feed_mgr_ = const_cast<coin2::feed::mgr::FeedSubsystemManager*>(feed_mgr);
    // set window size of moving average
    spot_book_processed_.ResizeWindow(60'000'000'000LL);
    spot_book_processed2_.ResizeWindow(60'000'000'000LL);
    fut_book_processed_.ResizeWindow(60'000'000'000LL);
    fut_book_processed2_.ResizeWindow(60'000'000'000LL);
    spot_trade_processed_.ResizeWindow(60'000'000'000LL);
    spot_trade_processed2_.ResizeWindow(60'000'000'000LL);
    fut_trade_processed_.ResizeWindow(60'000'000'000LL);
    fut_trade_processed2_.ResizeWindow(60'000'000'000LL);

    spdlog::set_pattern("%L%m%d %H:%M:%S.%f %P %s:%#] %v");
    CHECK_GT(timestamp_override, 0);

    // read lm_strat_config
    std::ifstream in(lm_strat_json);
    CHECK_THROW(in.is_open());
    std::string str((std::istreambuf_iterator<char>(in)),
                    std::istreambuf_iterator<char>());
    google::protobuf::util::JsonStringToMessage(str, &lm_strat_config_);
    presto::quant::feature::CutCustomData(lm_strat_config_.mutable_custom_data(), timestamp_override);
    LOG(INFO) << fmt::format("lm_strat_json: {}", lm_strat_config_.DebugString());
    global_feed_id_ = 0L;
    in.close();

    bool use_kline = false;
    if (!kline_json.empty()) {
      in.open(kline_json);
      std::string str2((std::istreambuf_iterator<char>(in)),
                       std::istreambuf_iterator<char>());
      auto status = google::protobuf::util::JsonStringToMessage(str2, kline_info.get());
      CHECK(status.ok()) << fmt::format("invalid proto json: {}", str2.substr(0, 10000));
      in.close();
      use_kline = true;

      if (lm_strat_config_.has_custom_data()) {
        *kline_info->mutable_custom_data() = lm_strat_config_.custom_data();
      }
    }

    int i = 0;
    for (auto product_config : lm_strat_config_.product_configs()) {
      auto mea = MarketExchangeApi::FromString(product_config.mea());
      std::string profile_root = product_config.profile_root();
      std::string profile_zip = fmt::format(
          "{0}/{1}.zip",
          profile_root,
          product_config.profile_name());
      CHECK(product_config.profile_absolute_path().empty());

      if (!product_config.xylm_json().empty()) {
        xylm_proto_.reset(new ::fastfeature::XylmProto());
        google::protobuf::util::JsonStringToMessage(
            product_config.xylm_json(),
            xylm_proto_.get());
      }

      // uncomment below line when you want to test fatsim functionality
      // ++i;
      auto key = std::make_tuple(
          profile_zip,
          product_config.linear_alpha_file(),
          i);

      std::vector<presto::Mep> meps;
      for (auto product : feed_mgr->GetProducts()) {
        meps.push_back(presto::Mep{
            MarketType_Name(product->market()),
            ExchangeType_Name(product->exchange()),
            product->relative_norm()
            });
      }

      int64_t rank_ge;
      int64_t rank_lt;
      if (GetFocusSymbolsVuRankGeLt(
            GetFSpecProtoFromZip(profile_zip, product_config.linear_alpha_file()),
            &rank_ge,
            &rank_lt)) {
        auto feature_spec_pb = GetFSpecProtoFromZip(profile_zip, product_config.linear_alpha_file());
        bool require_quote_matching = feature_spec_pb.focus_vu_require_quote_matching();
        bool exclude_noref_symbols = feature_spec_pb.focus_vu_exclude_noref_symbols();
        LOG(INFO) << "require_quote_matching=" << require_quote_matching;
        meps = BanFocusRankAndGetAllowlist(
            feed_mgr_,
            lm_strat_config_.volume_info(),
            timestamp_override + nanosecs_warmedup,
            rank_ge,
            rank_lt,
            &clear_meps_,
            product_config.mutable_focus_config(),
            require_quote_matching,
            exclude_noref_symbols);
      }
      if (lm_cache_.count(key) == 0) {
        lm_cache_[key] = LinearModelWrapper{
            global_feed_id_,
            std::unique_ptr<LinearModel>(),
            LinearModelDelegateProto()};
        if (use_kline) {
          lm_cache_.at(key).linear_model = CreateNewLinearModelFromArchiveWithKline(
              profile_zip,
              product_config.linear_alpha_file(),
              &lm_cache_.at(key).delegate,
              kline_info.get(),
              xylm_proto_.get(),
              meps);
        } else {
          lm_cache_.at(key).linear_model = CreateNewLinearModelFromArchive(
              profile_zip,
              product_config.linear_alpha_file(),
              &lm_cache_.at(key).delegate,
              xylm_proto_.get(),
              meps);
        }
      }
      auto& lmwrapper = lm_cache_.at(key);
      // logic and sub-strat for 1 target symbol
      lm_strats_.push_back(std::make_unique<ProductLmStrategy>(
          lmwrapper.linear_model.get(),
          &lmwrapper.lm_feed_id,
          &global_feed_id_,
          &lmwrapper.delegate,
          mea,
          product_config.symbol_str(),
          product_config,
          &product_cache_,
          timestamp_override,
          simulation_,
          lm_strat_config_.volume_info(),
          clear_meps_,
          use_kline? kline_info : nullptr));
      lm_strats_.back()->GetLinearModelLogicMutable()->SetFeedMgr(feed_mgr_);
    }
  }

  void Init(::coin2::app::Driver* driver) override {
    driver_ = driver;
    auto curr_time = GetCurrentTimestamp();
    for (auto& [name, oe] : driver->strategy()->order_executors(0L)) {
      (void)name;
      LOG(INFO) << oe->config().DebugString();
    }

    strat_reporter_.reset(new util::StrategyReporter2(driver->strategy()->config()));

    if (driver->config().has_alarm()) {
      alarm_.reset(new util::AlarmManager(driver->config().alarm(), curr_time));
    }

    for (auto product_config : lm_strat_config_.product_configs()) {
      for (auto& [key, lm_strat] : lm_cache_) {
        (void)key;
        AddClearMepsToLinearModel(lm_strat.linear_model.get(), clear_meps_);
      }
    }

    for (auto& lm_strat : lm_strats_) {
      lm_strat->set_strategy_logger(strat_reporter_->GetMutableStrategyLogger());
      lm_strat->Init(driver);
      if (alarm_) lm_strat->RegisterAlarm(alarm_.get());
    }
    std::string strat_group = driver_->strategy()->config().strategy_group();
    is_focus_ = StringStartsWith(strat_group, "lm-focus")
      || strat_group == "pilot_fund"
      || strat_group == "digital_fund"
      || strat_group == "lm-focus-midfreq"
      || strat_group == "midfreq_longshort"
      || strat_group == "midfreq_portfolio"
      || strat_group == "midfreq_longonly"
      || strat_group == "lowfreq_longshort"
      || strat_group == "midfreq_hedged"
      || strat_group == "midfreq_double"
      || strat_group == "midfreq_volume";
    logging_congestion_ = strat_group == "midfreq_longshort"
      || strat_group == "lm-focus-midfreq"
      || strat_group == "midfreq_longonly"
      || strat_group == "lowfreq_longshort"
      || strat_group == "midfreq_hedged"
      || strat_group == "midfreq_double"
      || strat_group == "midfreq_volume";
  }

  void onCleanup(::coin2::app::Driver* driver) override {
    for (auto& [name, oe] : driver->strategy()->order_executors(0L)) {
      (void)name;
      oe->ForceCancelAllWorkingOrders();
      submit_cnt_ += oe->GetSubmitCnt();
    }
  }

  int64_t GetSubmitCnt() const {
    return submit_cnt_;
  }

  void onTradeFeed(const FeedUpdate& upd) override {
    ++global_feed_id_;
    const ::fastfeature::FeedMessage* feed_msg_reuse = nullptr;
    for (auto&& lm_strat : lm_strats_) {
      if (feed_msg_reuse != nullptr) {
        lm_strat->onFeatureFeed(upd, feed_msg_reuse, simulation_);
      } else {
        lm_strat->onTradeFeed(upd);
        if (lm_strat->converter->feed_converted) {
          feed_msg_reuse = &lm_strat->converter->last_feed_msg;
        }
      }
    }

    if (logging_congestion_) {
      // mark packet timestamp
      int64_t raw_ts = upd.ts().raw_rx_timestamp();
      int64_t main_ts = upd.ts().main_tx_timestamp();
      int64_t main2_ts = GetCurrentTimestamp();
      if (upd.market() == MarketType::Spot) {
        spot_trade_processed_.Update(raw_ts, main_ts - raw_ts);
        spot_trade_processed2_.Update(raw_ts, main2_ts - raw_ts);
      } else if (upd.market() == MarketType::Futures) {
        fut_trade_processed_.Update(raw_ts, main_ts - raw_ts);
        fut_trade_processed2_.Update(raw_ts, main2_ts - raw_ts);
      }
    }
  }

  void onBookFeed(const FeedUpdate& upd) override {
    ++global_feed_id_;
    const ::fastfeature::FeedMessage* feed_msg_reuse = nullptr;
    last_book_time_ = upd.timestamp();
    for (auto&& lm_strat : lm_strats_) {
      if (feed_msg_reuse != nullptr) {
        lm_strat->onFeatureFeed(upd, feed_msg_reuse, simulation_);
      } else {
        lm_strat->onBookFeed(upd);
        if (lm_strat->converter->feed_converted) {
          feed_msg_reuse = &lm_strat->converter->last_feed_msg;
        }
      }
    }
    if (alarm_) {
      alarm_->onBookFeed(upd);
    }

    if (logging_congestion_) {
      // mark packet timestamp
      int64_t raw_ts = upd.ts().raw_rx_timestamp();
      int64_t main_ts = upd.ts().main_tx_timestamp();
      int64_t main2_ts = GetCurrentTimestamp();
      if (upd.market() == MarketType::Spot) {
        spot_book_processed_.Update(raw_ts, main_ts - raw_ts);
        spot_book_processed2_.Update(raw_ts, main2_ts - main_ts);
      } else if (upd.market() == MarketType::Futures) {
        fut_book_processed_.Update(raw_ts, main_ts - raw_ts);
        fut_book_processed2_.Update(raw_ts, main2_ts - main_ts);
      }

      // logging
      SPDLOG_INFO_EVERY_NS(
          60'000'000'000LL,
          raw_ts,
          "[processed time] spot book: {} / {} ms ({}), trade: {} / {} ms ({}), "
          "futures book: {} / {} ms ({}), trade: {} / {} ms ({})",
          spot_book_processed_.sum() * 1e-6,
          spot_book_processed2_.sum() * 1e-6,
          spot_book_processed_.num_elements(),
          spot_trade_processed_.sum() * 1e-6,
          spot_trade_processed2_.sum() * 1e-6,
          spot_trade_processed_.num_elements(),
          fut_book_processed_.sum() * 1e-6,
          fut_book_processed2_.sum() * 1e-6,
          fut_book_processed_.num_elements(),
          fut_trade_processed_.sum() * 1e-6,
          fut_trade_processed2_.sum() * 1e-6,
          fut_trade_processed_.num_elements());
    }
  }

  void onStatusFeed(const FeedUpdate& upd) {
    ++global_feed_id_;
    const ::fastfeature::FeedMessage* feed_msg_reuse = nullptr;
    for (auto&& lm_strat : lm_strats_) {
      if (feed_msg_reuse != nullptr) {
        lm_strat->onFeatureFeed(upd, feed_msg_reuse, simulation_);
      } else {
        lm_strat->onStatusFeed(upd);
        if (lm_strat->converter->feed_converted) {
          feed_msg_reuse = &lm_strat->converter->last_feed_msg;
        }
      }
    }
  }


  void onLiquidationFeed(const FeedUpdate& upd) override {
    onStatusFeed(upd);
  }

  void onFundingRateFeed(const FeedUpdate& upd) override {
    onStatusFeed(upd);
  }

  void onIndexFeed(const FeedUpdate& upd) override {
    onStatusFeed(upd);
  }

  void onOpenInterestFeed(const FeedUpdate& upd) override {
    onStatusFeed(upd);
  }

  void onTopLongShortAccountRatioFeed(const exchange::base::feed::FeedUpdate& upd) override {
    onStatusFeed(upd);
  }

  void onTopLongShortPositionRatioFeed(const exchange::base::feed::FeedUpdate& upd) override {
    onStatusFeed(upd);
  }

  void onGlobalLongShortAccountRatioFeed(const exchange::base::feed::FeedUpdate& upd) override {
    onStatusFeed(upd);
  }

  void onOrderLog(const coin::proto::OrderGatewayLog& order_log) override {
    if (strat_reporter_) {
      strat_reporter_->GetMutableStrategyLogger()
          ->WriteOrderLog(order_log, true, GetCurrentTimestamp());
    }
  }

  void onAccountInfo(const coin2::exchange::base::order::OrderUpdate& upd) override {
    if (strat_reporter_) {
      if (last_mtm_update_time_ + 1e9L < last_book_time_) {
        last_mtm_update_time_ = last_book_time_;
        for (auto& lm_strat : lm_strats_) {
          lm_strat->UpdateMidp(strat_reporter_.get());
        }
      }
      for (auto& lm_strat : lm_strats_) {
        lm_strat->UpdateAccountInfo(upd);
      }
      strat_reporter_->onAccountInfo(upd);
    }
  }

  void onAccountOrder(const coin2::exchange::base::order::OrderUpdate& upd) override {
    if (upd.product_order_info().event().type() == coin::proto::OrderEvent::ORDER_FILLED) {
      OnFill(upd);
    }
  }

  void OnFill(const coin2::exchange::base::order::OrderUpdate& upd) {
    const auto& key = GetProductKey(&upd.product_order_info().event());
    bool found = false;
    if (is_focus_) {
      found = true;
      for (auto& lm_strat : lm_strats_) {
        lm_strat->OnFill(
            upd.product_order_info().event().event_time(),
            upd.product_order_info().event().trade_side(),
            upd.product_order_info().event().fill_price(),
            upd.product_order_info().event().fill_qty());
        lm_strat->UpdateAccountOrder(upd);
      }
    } else {
      for (auto& lm_strat : lm_strats_) {
        auto* target_product = lm_strat->GetTargetProduct();
        if (!target_product) continue;
        if (GetProductKey(target_product) == key) {
          lm_strat->OnFill(
              upd.product_order_info().event().event_time(),
              upd.product_order_info().event().trade_side(),
              upd.product_order_info().event().fill_price(),
              upd.product_order_info().event().fill_qty());
          lm_strat->UpdateAccountOrder(upd);
          found = true;
        }
      }
    }
    if (found == false) {
      LOG(ERROR) << fmt::format("Received unexpected fills {}", key);
      unexpected_fills_cnt_++;
      CHECK(unexpected_fills_cnt_ < 10);
    }
  }

 private:
  const bool simulation_;
  bool is_focus_ = false;
  bool logging_congestion_ = false;
  int64_t last_mtm_update_time_;
  int64_t last_book_time_;
  int64_t global_feed_id_;
  LinearModelCache lm_cache_;
  std::vector<std::unique_ptr<IProduct>> product_owns_;
  std::vector<std::unique_ptr<ProductLmStrategy>> lm_strats_;
  std::vector<presto::Mep> clear_meps_;
  coin2::feed::mgr::FeedSubsystemManager* feed_mgr_ = nullptr;
  LmStrategyConfig lm_strat_config_;

  // product holder cache
  ::coin2::exchange::base::symbology::ProductEncyclopedia product_cache_;
  std::unique_ptr<util::StrategyReporter2> strat_reporter_;
  std::unique_ptr<util::ConfigReporter> config_reporter_;
  std::unique_ptr<util::AlarmManager> alarm_;
  ::coin2::app::Driver* driver_ = nullptr;

  std::unique_ptr<::fastfeature::XylmProto> xylm_proto_;

  TimeWindowMovingAverage<int64_t> spot_book_processed_;
  TimeWindowMovingAverage<int64_t> spot_book_processed2_;
  TimeWindowMovingAverage<int64_t> fut_book_processed_;
  TimeWindowMovingAverage<int64_t> fut_book_processed2_;
  TimeWindowMovingAverage<int64_t> spot_trade_processed_;
  TimeWindowMovingAverage<int64_t> spot_trade_processed2_;
  TimeWindowMovingAverage<int64_t> fut_trade_processed_;
  TimeWindowMovingAverage<int64_t> fut_trade_processed2_;
  std::vector<presto::Mep> feature_ban_symbols_;
  int64_t submit_cnt_ = 0L;
  int64_t unexpected_fills_cnt_ = 0L;
};  // namespace impl

}  // namespace impl
using impl::LmStrategy;
}  // namespace coin2::strategy::linear_model
